/*
 * 文件名：DataSendMonitorThread.java
 * 版权：亚信联创版权所有
 * 描述：TODO
 * 修改人：yuanze
 * 修改时间：2014年11月14日
 * 修改内容：TODO
 */
package com.ailk.bigdata.etl.realstream.server.thread;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ailk.bigdata.etl.common.server.constant.Constant;
import com.ailk.bigdata.etl.realstream.server.model.RealDataDispatcher;
import com.ailk.bigdata.etl.realstream.server.model.RealStreamUnit;

/**
 * @description 数据发送监控线程 数据如果长时间未处理 会溢出成文件
 * @author [yuanze]
 * @version [版本号,2014年11月14日]
 * @see [相关类/方法]
 * @since [产品/模块版本]
 */
public class DataSendMonitorThread implements Runnable
{
	private static Logger logger = LoggerFactory.getLogger(DataSendMonitorThread.class);

	private RealDataDispatcher dispatcher;
	private DataSendThread dataSendThread;

	public DataSendMonitorThread(RealDataDispatcher dispatcher)
	{
		this.dispatcher = dispatcher;
	}

	/**
	 * @see java.lang.Thread#run()
	 */
	@Override
	public void run()
	{
		RealStreamUnit unit = dispatcher.getRealStreamUnit();
		int minute = unit.getTransWaitMinute();

		while (true)
		{
			// 被动数据模式监控
			if (Constant.STREAM_TRANS_PASSIVE_MODE == unit.getStreamTransMode())
			{
				logger.info("被动模式启动");
				passiveModeMonitor();
			}
			else if (Constant.STREAM_TRANS_INITIATIVE_MODE == unit.getStreamTransMode())
			{
				logger.info("主动模式启动");
				initiativeModeMonitor();
			}
			try
			{
				Thread.sleep(minute * 60 * 1000);
			}
			catch (InterruptedException e)
			{
				logger.error("单元接口[" + unit.getOnlyUnitCode() + "]数据监控异常", e);
			}
		}
	}

	private void passiveModeMonitor()
	{
		// logger.info("[" + dispatcher.getRealStreamUnit().getOnlyUnitCode() +
		// "]被动发送线程监听启动............................");
		// boolean channelIsNormal = false;
		// for (Map.Entry<String, Channel> channelKey :
		// dispatcher.getSEND_CHANNEL_CACHE().entrySet())
		// {
		// Channel channel = channelKey.getValue();
		// if (!channel.isActive() || !channel.isWritable())
		// {
		// dispatcher.getSEND_CHANNEL_CACHE().remove(channelKey.getKey());
		// }
		// else
		// {
		// channelIsNormal = true;
		// dispatcher.setSendable(true);
		// }
		// }
		// dispatcher.setSendable(channelIsNormal);
		// // 启动该数据单元数据发送线程(线程的关闭由发送线程自己关闭)
		// if (dispatcher.isSendable() && !dispatcher.isRunning())
		// {
		// logger.info("[" + dispatcher.getRealStreamUnit().getOnlyUnitCode() +
		// "]被动发送线程监听到数据可发送............................");
		// DataFileBackupService.getBufferedPack(dispatcher);
		// dataSendThread = new DataSendThread(dispatcher);
		// logger.info("单元接口[" +
		// dispatcher.getRealStreamUnit().getOnlyUnitCode() +
		// "]发送消息线程准备启动.........");
		// ThreadPool.getTheadPool(RealStreamTool.THREAD_DISPATCHER_POOL_NAME).runTask(dataSendThread);
		// logger.info("被动发送线程监听到数据可发送.[" +
		// dispatcher.getRealStreamUnit().getOnlyUnitCode() +
		// "].end...........................");
		// }
		// // 监测到channel不正常且处理线程未关闭(强制关闭)
		// if (!dispatcher.isSendable())
		// {
		// logger.info("[" + dispatcher.getRealStreamUnit().getOnlyUnitCode() +
		// "]被动发送线程监听到数据不可发送缓存到文件中............................");
		// if (dispatcher.isRunning())
		// {
		// if (dataSendThread != null && dataSendThread.isAlive())
		// {
		// dataSendThread.interrupt();
		// }
		// dispatcher.setRunning(false);
		// }
		// // 开始将数据同步到文件
		// DataFileBackupService.backToFile(dispatcher);
		// logger.info("被动发送线程监听到数据不可发送缓存到文件中.[" +
		// dispatcher.getRealStreamUnit().getOnlyUnitCode() +
		// "].end...........................");
		// }
	}

	private void initiativeModeMonitor()
	{
		// boolean channelIsNormal = true;
		// // 获得主动推送远程地址 首次进入建立channel连接
		// List<SocketInfo> socketInfoList =
		// dispatcher.getRealStreamUnit().getSocketInfo();
		// for (SocketInfo socketInfo : socketInfoList)
		// {
		// String key = socketInfo.getSocketIp() + socketInfo.getSocketPort();
		// Channel channel = dispatcher.getSEND_CHANNEL_CACHE().get(key);
		// if (channel == null || !channel.isActive() || !channel.isWritable())
		// {
		// Channel newChannel = new
		// StreamDataInitiativeSendClient(socketInfo.getSocketIp(),
		// socketInfo.getSocketPort()).run();
		// if (newChannel != null)
		// {
		// dispatcher.getSEND_CHANNEL_CACHE().put(key, newChannel);
		// }
		// else
		// {
		// channelIsNormal = false;
		// }
		// }
		// }
		// dispatcher.setSendable(channelIsNormal);
		// // 启动该数据单元数据发送线程(线程的关闭由发送线程自己关闭)
		// if (dispatcher.isSendable() && !dispatcher.isRunning())
		// {
		// DataFileBackupService.getBufferedPack(dispatcher);
		// dataSendThread = new DataSendThread(dispatcher);
		// logger.info("单元接口[" +
		// dispatcher.getRealStreamUnit().getOnlyUnitCode() +
		// "]发送消息线程准备启动.........");
		// ThreadPool.getTheadPool(RealStreamTool.THREAD_DISPATCHER_POOL_NAME).runTask(dataSendThread);
		// }
		// // 监测到channel不正常且处理线程未关闭(强制关闭)
		// if (!dispatcher.isSendable())
		// {
		// if (dispatcher.isRunning())
		// {
		// if (dataSendThread != null && dataSendThread.isAlive())
		// {
		// dataSendThread.interrupt();
		// }
		// dispatcher.setRunning(false);
		// }
		// // 开始将数据同步到文件
		// DataFileBackupService.backToFile(dispatcher);
		// }
	}
}
